package service

import (
	"encoding/json"
	"fmt"
	"gitee.com/thomasgw/gms/common"
	"gitee.com/thomasgw/gms/outer"
	"github.com/oleiade/lane"
	"github.com/syndtr/goleveldb/leveldb"
	"github.com/syndtr/goleveldb/leveldb/opt"
	"github.com/syndtr/goleveldb/leveldb/util"
	"strconv"
	"sync/atomic"
	"time"
	"sync"
)

type Service struct {
	serverCh chan interface{}
	DB       *leveldb.DB
	eventq   *lane.Queue
	outerCom *outer.Component
	lock     *sync.Mutex
	gc       *GC
}

func NewService(serverCh chan interface{}) *Service {
	db, err := leveldb.OpenFile(common.DB_PATH, &opt.Options{})
	if err != nil {
		fmt.Println(err.Error())
		return nil
	} else {
		lock := new(sync.Mutex)
		srv := &Service{
			serverCh: serverCh,
			DB:       db,
			eventq:   lane.NewQueue(),
			lock:     lock,
			gc:       NewGC(db, lock),
		}
		srv.gc.initState()
		return srv
	}
}

func (service *Service) Start() {
	err := service.recover()
	if err != nil {
		fmt.Println("Cannot do recover, reason: %v, skip recover", err.Error())
	}
	service.gc.stateMachine.Event(EVENT_START, nil)
	go service.Listening()
	go service.Dispatching()
}

func (service *Service) Listening() {

	// todo: version1: for each item that the server sent out, write database then enqueue,
	// todo: which means write database is a sync process.

	for item := range service.serverCh {
		item, ok := item.(common.ServerEvent)
		if !ok {
			fmt.Println("Received an event which type is not ServerEvent, ignore it.")
		} else {
			key := append(common.EventPrefix, []byte(item.Timestamp)...)
			result := common.SerializeEvent(item)
			if result != nil {
				err := service.DB.Put(key, result, nil)
				if err != nil {
					fmt.Println("DB write error: ", err.Error())
					continue
				}
			} else {
				fmt.Println("ServerEvent is not in json format, invalid event content, ignore that.")
				continue
			}
			service.eventq.Enqueue(item)
		}
	}
}

func (service *Service) Dispatching() {
	itemCount := int32(0)
	for {
		time.Sleep(300 * time.Millisecond)
		if service.gc.stateMachine.Current() == STATE_NORMAL {
			if item := service.eventq.Head(); item != nil {
				e := item.(common.ServerEvent)
				ACK := service.Send(e)
				if ACK {
					dbKey := append(common.EventPrefix, []byte(e.Timestamp)...)
					service.updateDBCursor(dbKey)
					service.eventq.Pop()
					atomic.StoreInt32(&itemCount, 0)
				} else {
					atomic.AddInt32(&itemCount, 1)
				}
				if swapped := atomic.CompareAndSwapInt32(&itemCount, common.THRESHOLD, 0); swapped {
					failedItem := service.eventq.Pop()
					service.eventq.Enqueue(failedItem)
				}
			}
		}
	}
}

func (service *Service) updateDBCursor(dbKey []byte ) {
	service.lock.Lock()
	defer service.lock.Unlock()
	service.DB.Put(common.CursorKey, dbKey, nil)
}

func (service *Service) Send(item interface{}) bool {
	return service.outerCom.Receive(item)
}

func (service *Service) recover() error {
	cursor, err := service.DB.Get(common.CursorKey, nil)
	if err == leveldb.ErrNotFound {
		fmt.Println("Cursor not found, cannot do recover.")
		return err
	}
	timeStampNow := strconv.Itoa(int(time.Now().UnixNano()))
	ite := service.DB.NewIterator(&util.Range{
		Start: cursor,
		Limit: append(common.EventPrefix, []byte(timeStampNow)...),
	}, nil)
	for ite.Next() {
		s := make([]byte, len(ite.Value()))
		copy(s, ite.Value())
		event := &common.ServerEvent{}
		err = json.Unmarshal(s, event)
		if err != nil {
			fmt.Println("Data parse to event error")
			continue
		} else {
			fmt.Println("[RECOVER] Get recover content: ", event)
		}
		service.eventq.Enqueue(*event)
	}
	ite.Release()
	return nil
}

func (service *Service) Register() {
	service.outerCom = outer.NewComponent()
}

func (service *Service) GetOuterCom() *outer.Component {
	return service.outerCom
}
